package com.xingfly.event;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.xingfly.event.properties.RabbitProperties;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.support.converter.DefaultClassMapper;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by SuperS on 2019/11/19.
 *
 * @author SuperS
 */
@Configuration
public class MqConfiguration {

    private RabbitProperties rabbitProperties;

    public MqConfiguration(RabbitProperties rabbitProperties) {
        this.rabbitProperties = rabbitProperties;
    }

    @Bean
    public MessageConverter jsonMessageConverter(ObjectMapper objectMapper) {
        Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter(objectMapper);
        messageConverter.setClassMapper(classMapper());
        return messageConverter;
    }

    @Bean
    public DefaultClassMapper classMapper() {
        DefaultClassMapper classMapper = new DefaultClassMapper();
        classMapper.setTrustedPackages("*");
        return classMapper;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory,
                                                                               MessageConverter messageConverter) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(messageConverter);
        return factory;
    }

    /**
     * 发送方Exchange
     * alternate-exchange:如果消息不能根据RoutingKey路由到队列 就会路由到Alternate Exchange
     */
    @Bean
    public TopicExchange publishExchange() {
        Map<String, Object> args = new HashMap<>(1);
        args.put("alternate-exchange", rabbitProperties.getPublishDlx());
        return new TopicExchange(rabbitProperties.getPublishX(), true, false, args);
    }

    /**
     * "发送方DLX"，消息发送失败时传到该DLX
     */
    @Bean
    public TopicExchange publishDlx() {
        return new TopicExchange(rabbitProperties.getPublishDlx(), true, false, null);
    }

    /**
     * "发送方DLQ"，所有发到"发送DLX"的消息都将路由到该DLQ
     * x-queue-mode: 当该值设置为“lazy”时，消息将先保存到磁盘上、不放在内存中，当消费者开始消费的时候才加载到内存中。
     */
    @Bean
    public Queue publishDlq() {
        Map<String, Object> args = new HashMap<>(1);
        args.put("x-queue-mode", "lazy");
        return new Queue(rabbitProperties.getPublishDlq(), true, false, false, args);
    }

    /**
     * "发送方DLQ"绑定到"发送方DLX"
     */
    @Bean
    public Binding orderPublishDlqBinding() {
        return BindingBuilder.bind(publishDlq()).to(publishDlx()).with("#");
    }

    /**
     * 接收方的所有消息都发送到该"接收方Queue"
     * x-dead-letter-exchange: 声明死信交换机
     * x-overflow: (drop-head)当超过最大积压数量时会删除头部消息腾出空间存放新消息
     * x-max-length: 设置队列中的消息的最大条数
     * x-message-ttl: message 被 发到队列中后 在队列中的存活时间 超过时间即为已死
     */
    @Bean
    public Queue receiveQ() {
        Map<String, Object> args = new HashMap<>(4);
        args.put("x-dead-letter-exchange", rabbitProperties.getReceiveDlx());
        args.put("x-overflow", "drop-head");
        args.put("x-max-length", 300000);
        args.put("x-message-ttl", 24 * 60 * 60 * 1000);
        return new Queue(rabbitProperties.getReceiveQ(), true, false, false, args);
    }

    /**
     * "接收方DLX"，消息处理失败时传到该DLX
     */
    @Bean
    public TopicExchange receiveDlx() {
        return new TopicExchange(rabbitProperties.getReceiveDlx(), true, false, null);
    }

    /**
     * "接收方DLQ"，所有发到"接收DLX"的消息都将路由到该DLQ
     * x-queue-mode: 当该值设置为“lazy”时，消息将先保存到磁盘上、不放在内存中，当消费者开始消费的时候才加载到内存中。(lazy)
     */
    @Bean
    public Queue receiveDlq() {
        Map<String, Object> args = new HashMap<>(1);
        args.put("x-queue-mode", "lazy");
        return new Queue(rabbitProperties.getReceiveDlq(), true, false, false, args);
    }

    /**
     * "接收方DLQ"绑定到"接收方DLX"
     */
    @Bean
    public Binding receiveDlqBinding() {
        return BindingBuilder.bind(receiveDlq()).to(receiveDlx()).with("#");
    }

    /**
     * "接收方恢复Exchange"，用于手动将"接收方DLQ"中的消息发到该DLX进行重试
     */
    @Bean
    public TopicExchange receiveRecoverExchange() {
        return new TopicExchange(rabbitProperties.getReceiveRecoverX(), true, false, null);
    }

    /**
     * "接收方Queue"绑定到"接收方恢复Exchange"
     */
    @Bean
    public Binding receiveRecoverBinding() {
        return BindingBuilder.bind(receiveQ()).to(receiveRecoverExchange()).with("#");
    }


}
